package kafkaserver

import (
	"context"
	"core/kafka"
	"core/server/baseserver"
	"github.com/dubbogo/gost/log/logger"
	kafkago "github.com/segmentio/kafka-go"
	"strings"
	"time"
)

type ConsumerServer struct {
	Name      string
	Topic     string
	GroupID   string
	Do        func(message kafkago.Message) error
	closeChan chan int32
}

func (p *ConsumerServer) Start() {
	p.closeChan = make(chan int32, 1)
	baseserver.AddServer(p)
	logger.Infof("接收程序：%v topic:%v  启动", p.Name, p.Topic)
	go p.Run()
}
func (p *ConsumerServer) Close() {
	p.closeChan <- 1

}

func (p *ConsumerServer) GetKafkaReader(topic, groupID string) *kafkago.Reader {
	brokers := strings.Split(kafka.Config.Dsn, ",")
	return kafkago.NewReader(kafkago.ReaderConfig{
		Brokers:  brokers,
		GroupID:  groupID,
		Topic:    topic,
		MinBytes: 10e3, // 10KB
		MaxBytes: 10e6, // 10MB
		MaxWait:  time.Millisecond * 100,
	})
}

func (p *ConsumerServer) Run() {
	reader := p.GetKafkaReader(p.Topic, p.GroupID)

	defer reader.Close()
	for {
		select {
		case <-p.closeChan:
			logger.Infof("接收程序：%v topic:%v  关闭", p.Name, p.Topic)
			break
		default:
			m, err := reader.ReadMessage(context.Background())
			if err != nil {
				logger.Warnf("读取kafka %v %v失败", p.Topic, p.GroupID)
			} else {
				if p.Do != nil {
					err = p.Do(m)
				}
			}
		}
	}
}
